/*
 * jETeL/CloverETL - Java based ETL application framework.
 * Copyright (c) Javlin, a.s. (info@cloveretl.com)
 *  
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 */
package org.jetel.data.tape;

import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jetel.data.DataRecord;
import org.jetel.data.Defaults;
import org.jetel.exception.JetelRuntimeException;
import org.jetel.graph.ContextProvider;
import org.jetel.graph.runtime.IAuthorityProxy;
import org.jetel.util.bytes.ByteBufferUtils;
import org.jetel.util.bytes.CloverBuffer;


/**
 * Data structure for storing records in chunks on local disk (in temporary file).<br>
 * When the data object is closed - the physical representation (the file) is removed - i.e.
 * no permanent storage is performed.<br>
 * Each chunk registers how many records is stored in it. The number
 * of chunks on tape(in file) is not limited. Records can be read only
 * sequentially, but particular chunk can be selected.<br><br>
 * <i>Usage:</i><br>
 * <code>
 * tape=new DataRecordTape();<br>
 * tape.open();<br>
 * tape.addDataChunk();<br>
 * ..loop.. tape.put(..data..);<br>
 * tape.addDataChunk();<br>
 * ..loop.. tape.put(..data..);<br>
 * tape.rewind();<br>
 * ..loop..tape.get();<br>
 * tape.nextChunk();<br>
 * ..loop..tape.get();<br>
 * tape.close();<br>
 * </code>
 * 
 * @author david
 * @since  20.1.2005
 *
 */
public class DataRecordTape implements Closeable {

    private RandomAccessFile tmpRandomAccessFile;
    private FileChannel tmpFileChannel;
	private File tmpFile;
	private String tmpFileName;

    private boolean deleteOnExit;
    private boolean deleteOnStart;
    
	private List<DataChunk> dataChunks;
	
	private CloverBuffer dataBuffer;

	private DataChunk currentDataChunk;
	private int currentDataChunkIndex;
	private boolean updatedAfterFlush;
    
	// size of BUFFER - used for push & shift operations
	private final static int DEFAULT_BUFFER_SIZE = Defaults.Record.RECORDS_BUFFER_SIZE; 

	// prefix of temporary file generated by system	
	private final static String TMP_FILE_PREFIX = "fbufclv";
	//	 suffix of temporary file generated by system
	private final static String TMP_FILE_SUFFIX = ".tmp";
	private final static String TMP_FILE_MODE = "rw";

	static Log logger = LogFactory.getLog(DataRecordTape.class);

	/**
	 *  Constructor for the DataRecordTape object
	 *
	 *@param  tmpFileName      Name of the temp file or NULL (the system default temp directory and name will be used)
	 *@param  dataBufferSize   The size of internal in memory buffer.
     *                          If smaller than DEFAULT_BUFFER_SIZE, then default is used
	 */
	public DataRecordTape(String tmpFileName, int dataBufferSize, boolean deleteOnStart, boolean deleteOnExit) {
		this.tmpFileName = tmpFileName;
        this.deleteOnStart = deleteOnStart;
        this.deleteOnExit = deleteOnExit;
		dataChunks = new ArrayList<DataChunk>();
		dataBuffer = CloverBuffer.allocateDirect(dataBufferSize > DEFAULT_BUFFER_SIZE ? dataBufferSize : DEFAULT_BUFFER_SIZE);
	}

    /**
     * Constructor.
     * @param tmpFileName
     * @param deleteOnExit
     */
    public DataRecordTape(String tmpFileName, boolean deleteOnStart, boolean deleteOnExit) {
        this(tmpFileName,DEFAULT_BUFFER_SIZE, deleteOnStart, deleteOnExit);
    }

	/**
	 *  Constructor for the DataRecordTape object
	 *
     *@param  tmpFileName      Name of the temp file or NULL (the system default temp directory and name will be used)
	 */
	public DataRecordTape(String tmpFileName) {
		this(tmpFileName,DEFAULT_BUFFER_SIZE, true, true);
	}

	/**
	 * Constructor for DataRecordTape - all parameters defaulted.
	 */
	public DataRecordTape(){
	    this(null,DEFAULT_BUFFER_SIZE, true, true);
	}
	
	/**
	 * Opens buffer, creates temporary file.
	 * @throws IOException
	 */
	public void open() throws IOException {
		this.open(-1);
	}

	/**
	 *  Opens buffer, creates temporary file.
	 *
	 *@exception  IOException  Description of Exception
	 *@since                   September 17, 2002
	 */
	public void open(int allocationHint) throws IOException {
        if(tmpFileName == null)
			try {
				tmpFile = IAuthorityProxy.getAuthorityProxy(ContextProvider.getGraph()).newTempFile(TMP_FILE_PREFIX, TMP_FILE_SUFFIX, allocationHint);
			} catch (Exception e) {
				throw new IOException("Creating tmp file (prefix: " + TMP_FILE_PREFIX + ", suffix: " + TMP_FILE_SUFFIX + ") failed.", e);
			} 
		else {
            tmpFile = new File(tmpFileName);
            if(deleteOnStart && tmpFile.exists()) {
                if (!tmpFile.delete()) {
                    throw new IOException("Can't delete TMP file: " + tmpFile.getAbsoluteFile());
                }
            }
            if(!deleteOnStart && !tmpFile.exists()) {
                throw new IOException("Temp file does not exist.");
            }
        }
        
		// we want the temp file be deleted on exit
        tmpRandomAccessFile = new RandomAccessFile(tmpFile, TMP_FILE_MODE);
        tmpFileChannel = tmpRandomAccessFile.getChannel();
       	currentDataChunkIndex=-1;
		currentDataChunk=null;
	}


	/**
	 *  Closes buffer, removes temporary file (is exists)
	 *
	 *@exception  IOException  Description of Exception
	 * @throws InterruptedException 
	 *@since                   September 17, 2002
	 */
	@Override
	public void close() throws IOException {
        if(deleteOnExit) {
            clear();
            closeTmpFile();
            if (!tmpFile.delete()) {
                throw new IOException("Can't delete TMP file: " + tmpFile.getAbsoluteFile());
            }
        } else {
        	closeTmpFile();
        }
	}

	/**
	 * This special file closing tries to be sure the temporary file is really closed.
	 */
	private void closeTmpFile() throws IOException {
		if (tmpFileChannel != null) {
			tmpFileChannel.close();
		} else if (tmpRandomAccessFile != null) {
			tmpRandomAccessFile.close();
		}
	}
	
	/**
	 * Flushes tape content to disk.
	 * 
	 * @throws IOException
	 * @throws InterruptedException 
	 */
	public void flush(boolean force) throws IOException, InterruptedException {
		try {
		    dataBuffer.flip();
		    tmpFileChannel.write(dataBuffer.buf());
		    dataBuffer.clear();
		    //currentDataChunk.flushBuffer();
		    if (force){
		        tmpFileChannel.force(true);
		    }
		    updatedAfterFlush = false;
		} catch (ClosedChannelException e) {
			throw new InterruptedException();
		}
	}

	/**
	 * <p>Rewinds the buffer and makes first chunk active. Next get operation returns first record stored in
	 * first chunk. Allows preventing flush (see <code>doFlush</code> param.) that block concurrency in some situations.</p>
	 * 
	 * <p>Flush shall be done if there were some data written to the tape.
	 * 
	 * @param doFlush <code>true</code> if flush shall be done,<br/><code>false</code> otherwise
	 * @throws InterruptedException
	 * @throws IOException
	 * 
	 * @since September 19, 2002
	 */
	public void rewind(boolean doFlush) throws InterruptedException, IOException {
		
		if (updatedAfterFlush && !doFlush) {
			throw new IllegalStateException("not doing flush is allowed only if no data were written to tape since last flush");
		}
	    if (dataChunks.size()==0){
	        return;
	    }

	    try {
	    	if (doFlush) {
	    		flush(true);
	    	}
	    	tmpFileChannel.position(0);
	    	
		    currentDataChunkIndex=0;
		    currentDataChunk = dataChunks.get(0);
		    currentDataChunk.rewind();
	    } catch (ClosedChannelException e) {
	    	throw new InterruptedException();
	    }
	}
	

	/**
	 * @throws IOException 
	 * @throws InterruptedException 
	 * 
	 */
	public void rewind() throws InterruptedException, IOException {
		rewind(true);
	}

	/**
	 * Clears the internal buffer.
	 * <p>
	 * This method should be used before storing additional data records after all data
	 * records stored on the tape were read.
	 * <p>
	 * <b>Code fragment:</b>
	 * <pre>
	 *   tape.addDataChunk();
	 *   REPEAT tape.put(record);
	 *
	 *   tape.rewind();
	 *   REPEAT tape.get(record);
	 *
	 *   tape.clearBuffer();
	 *   REPEAT tape.put(record);
	 * </pre>
	 */
	public void clearBuffer() {
		dataBuffer.clear();
	}

	/**
	 *  Clears the tape. All DataChunks are destroyed. Underlying 
	 * tmp file is truncated.
	 * @throws InterruptedException 
	 */
	public void clear() throws IOException {
		dataChunks.clear();
		if( tmpFileChannel.isOpen() ){
			tmpFileChannel.truncate(0);
			tmpFileChannel.position(0);
		}
		currentDataChunkIndex=-1;
		currentDataChunk=null;
		updatedAfterFlush = false;
	}

	/**
	 * Adds new data chunk to the tape and opens it.
	 * @return true if successful, otherwise false
	 * @throws InterruptedException 
	 * @throws IOException 
	 */
	public void addDataChunk() throws InterruptedException, IOException {
	    if (currentDataChunk==null){
            // add new data chunk
            DataChunk chunk=new DataChunk(tmpFileChannel,dataBuffer);
            dataChunks.add(chunk);
            currentDataChunkIndex=0;
            currentDataChunk=chunk;
	    } else {
	    	try {
	            // set file position to the end of file
	        	flush(true);
	            tmpFileChannel.position(tmpFileChannel.size());
	            // add new data chunk
	            DataChunk chunk=new DataChunk(tmpFileChannel,dataBuffer);
	            dataChunks.add(chunk);
	            currentDataChunkIndex++;
	            currentDataChunk=chunk;
	    	} catch (ClosedChannelException e) {
	    		throw new InterruptedException();
	    	}
	    }
	    dataBuffer.clear();
	}

	/**
	 * Sets next data chunk active. Next get() operation will return first record
	 * from the newly activated chunk. This method must be called after rewind()
	 * method, otherwise the result is not guaranteed.
	 * 
	 * @return true if next data chunk has been activated, otherwise false
	 * @throws InterruptedException 
	 * @throws IOException 
	 * @throws IOException 
	 */
	public boolean nextDataChunk() throws InterruptedException, IOException {
	    if (currentDataChunkIndex!=-1 && currentDataChunkIndex+1 < dataChunks.size()){
	        currentDataChunkIndex++;
	        currentDataChunk = dataChunks.get(currentDataChunkIndex);
	        currentDataChunk.rewind();
		        
	        return true;
	    }else{
	        currentDataChunkIndex=-1;
			currentDataChunk=null;
	        return false;
	    }
	}
	
	
	public boolean setDataChunk(int order) throws InterruptedException, IOException {
	    if (order<dataChunks.size()){
	        currentDataChunk = dataChunks.get(order);
	        currentDataChunkIndex=order;
			updatedAfterFlush = true;
	        currentDataChunk.rewind();
	        
	        return true;
	    }else{
	        currentDataChunkIndex=-1;
			currentDataChunk=null;
	        return false;
	    }
	    
	}
	
	/**
     * Returns number of chunks this tape contains.
     * 
	 * @return
	 * @since 1.2.2007
	 */
	public int getNumChunks(){
	    return dataChunks.size();
	}
    
    
    /**
     * For specified chunk number returns its
     * length (in bytes).
     * 
     * @param chunk
     * @return
     * @since 1.2.2007
     */
    public long getChunkLength(int chunk) {
        return (dataChunks.get(chunk)).getLength();
    }
    
    /**
     * For specified chunk number returns how
     * many records it contains.
     * 
     * @param chunk
     * @return
     * @since 1.2.2007
     */
    public int getChunkRecNum(int chunk) {
        return (dataChunks.get(chunk)).getNumRecords();
    }
    
    /**
	 * @return size of allocated memory in internal buffers (memory footprint)
     */
    public int getBufferSize() {
    	int result = dataBuffer.capacity();
    	
    	return result;
    }
	
	/**
	 * Stores data in current/active chunk. Must not be mixed with calls to
	 * get() method, otherwise the result is not guaranteed.
	 * 
	 * @param data buffer containig record's data
	 * @return
	 * @throws InterruptedException 
	 */
	public long put(CloverBuffer data) throws IOException, InterruptedException {
		if (currentDataChunk != null) {
	        long putResult = currentDataChunk.put(data);
			updatedAfterFlush = true;
			return putResult;
		} else {
	        throw new RuntimeException("No DataChunk has been created !");
	    }
	}

	/**
	 * @deprecated use {@link #put(CloverBuffer)} instead
	 */
	@Deprecated
	public long put(ByteBuffer data) throws IOException, InterruptedException {
		CloverBuffer wrappedBuffer = CloverBuffer.wrap(data);
		long result = put(wrappedBuffer);
		if (wrappedBuffer.buf() != data) {
			throw new JetelRuntimeException("Deprecated method invocation failed. Please use CloverBuffer instead of ByteBuffer.");
		}
		updatedAfterFlush = true;
		return result;
	}

	/**
	 * Stores data record in current/active chunk.
	 * 
	 * @param data
	 * @return
	 * @throws IOException
	 * @throws InterruptedException 
	 */
	public long put(DataRecord data) throws IOException, InterruptedException {
		if (currentDataChunk != null) {
	        long putResult = currentDataChunk.put(data);
			updatedAfterFlush = true;
			return putResult;
		} else {
	        throw new RuntimeException("No DataChunk has been created !");
	    }
	}
	
	/**
	 * Fills buffer passed as an argument with data read from current data chunk.
	 * Must not be mixed with put() method calls, otherwise the result is not guaranteed.<br>
	 * The normal way of using this method is:<br>
	 * while(get(data)){
	 *   ... do something ...
	 * }
	 * @param data	buffer into which store data
	 * @return	true if success, otherwise false (chunk contains no more data).
	 * @throws InterruptedException 
	 */
	public boolean get(CloverBuffer data) throws IOException, InterruptedException {
	    if (currentDataChunk!=null){
	        return currentDataChunk.get(data);
	    }else{
	        return false;
	    }
	}
	
    /**
     * Reads again previously read record.
     * 
     * @param data
     * @return
     * @throws IOException
     * @since 27.2.2007
     */
    public boolean reget(CloverBuffer data) {
        if (currentDataChunk!=null){
            return currentDataChunk.reget(data);
        }else{
            return false;
        }
    }
    
    /**
     * @deprecated use {@link #reget(CloverBuffer)} instead
     */
    @Deprecated
    public boolean reget(ByteBuffer data) {
		CloverBuffer wrappedBuffer = CloverBuffer.wrap(data);
		boolean result = reget(wrappedBuffer);
		if (wrappedBuffer.buf() != data) {
			throw new JetelRuntimeException("Deprecated method invocation failed. Please use CloverBuffer instead of ByteBuffer.");
		}
		return result;
    }
    
	/**
	 * Reads data record from current chunk
	 * 
	 * @param data
	 * @return
	 * @throws IOException
	 * @throws InterruptedException  
	 */
	public boolean get(DataRecord data) throws IOException, InterruptedException {
	    if (currentDataChunk!=null){
	        return currentDataChunk.get(data);
	    }else{
	        return false;
	    }
	}
    
    /**
     * Reads again previously read record.
     * 
     * @param data
     * @return
     * @throws IOException
     * @since 27.2.2007
     */
    public boolean reget(DataRecord data) {
        if (currentDataChunk!=null){
            return currentDataChunk.reget(data);
        }else{
            return false;
        }
    }

	/* Returns String containing short summary of chunks stored on tape.
	 * 
	 * (non-Javadoc)
	 * @see java.lang.Object#toString()
	 */
	@Override
	public String toString() {
	    StringBuffer buffer=new StringBuffer(160);
	    int index=0;
	    for (DataChunk dataChunk : dataChunks) {
	        buffer.append("Chunk #").append(index++);
	        buffer.append(dataChunk.toString());
	        buffer.append("\n");
	    }
	    return buffer.toString();
	}
	
	public void testConsistency() throws InterruptedException, IOException {
		CloverBuffer buffer=CloverBuffer.allocateDirect(2048);
	    logger.info("Testing consistency...");

		rewind(true);
		
	    for(int i=0;i<getNumChunks();i++){
	        int counter=0;
	        try{
	            while(get(buffer)){
	                counter++;
	                buffer.clear();
	            }
		        if(!nextDataChunk()) break;
	        }catch(Exception ex){
	            logger.error("Problem with chunk: "+i+" record "+counter);
	            ex.printStackTrace();
	        }
	    }
	    logger.info("OK");
	}
	
	/**
	 * Helper class for storing data chunks. 
	 * @author david
	 * @since  20.1.2005
	 *
	 */
	private static class DataChunk{
	    //	size of integer variable used to keep record length
        // this is the maximum size, can be between 1 & 4 bytes
	    private final static int LEN_SIZE_SPECIFIER = 5;
	    CloverBuffer dataBuffer;
	    FileChannel tmpFileChannel;
	    long offsetStart;
	    long length;
	    int recordsRead;
	    int nRecords;
	    boolean canRead;
        int recordSize;
	    
	    private DataChunk(FileChannel channel,CloverBuffer buffer) throws InterruptedException, IOException{
	        tmpFileChannel=channel;
	        canRead=false;
	        dataBuffer=buffer;
	        try{
	            offsetStart=channel.position();
	    	} catch (ClosedChannelException e) {
	    		throw new InterruptedException();
	    	}
	        length=0;
	        nRecords=recordsRead=0;
	        dataBuffer.clear();
	    }
	    
	    long getLength(){
	        return length;
	    }
	    
	    int getNumRecords(){
	        return nRecords;
	    }
	    
	    void rewind() throws IOException, InterruptedException {
	    	try {
		        tmpFileChannel.position(offsetStart);
		        canRead=true;
		        recordsRead=0;
		        dataBuffer.clear();
		        tmpFileChannel.read(dataBuffer.buf());
		        dataBuffer.flip();
	    	} catch (ClosedChannelException e) {
	    		throw new InterruptedException();
	    	}
	    }
	    
	    /**
		 *  Stores one data record into buffer / file.
		 *
		 *@param  recordBuffer             ByteBuffer containing record's data
		 *@exception  IOException  In case of IO failure
		 *@since                   September 17, 2002
		 *@return number of bytes in the chunk after saveing data
	     * @throws InterruptedException 
		 */
		long put(CloverBuffer recordBuffer) throws IOException, InterruptedException {
			recordSize = recordBuffer.remaining();
			final int lengthSize=ByteBufferUtils.lengthEncoded(recordSize);
			// check that internal buffer has enough space
			if ((recordSize + lengthSize) > dataBuffer.remaining()) {
				flushBuffer();
			}
			
		    ByteBufferUtils.encodeLength(dataBuffer, recordSize);
            dataBuffer.put(recordBuffer);
			
			length+=(recordSize+ lengthSize);
			nRecords++;
			return length;
		}
        
		 /**
		  * Stores one data record into buffer / file.
		  * 
		 * @param data	DataRecord to be stored
		 *@return number of bytes in the chunk after saving data
		 * @throws IOException
		 * @throws InterruptedException 
		 */
		long put(DataRecord data) throws IOException, InterruptedException {
			recordSize = data.getSizeSerialized();
			final int lengthSize = ByteBufferUtils.lengthEncoded(recordSize);
			// check that internal buffer has enough space
			if ((recordSize + lengthSize) > dataBuffer.remaining()) {
				flushBuffer();
			}

			ByteBufferUtils.encodeLength(dataBuffer, recordSize);
			data.serialize(dataBuffer);

			length += (recordSize + lengthSize);
			nRecords++;
			return length;
		}

		/**
		 *  Returns next record from the buffer - FIFO order.
		 *
		 *@param  recordBuffer             ByteBuffer into which store data
		 *@return                  ByteBuffer populated with record's data or NULL if
		 *      no more record can be retrieved
		 *@exception  IOException  Description of Exception
		 * @throws InterruptedException 
		 *@since                   September 17, 2002
		 */
		 boolean get(CloverBuffer recordBuffer) throws IOException, InterruptedException {
			if(!canRead){
				throw new IOException("Buffer has not been rewind !");
			}
			
			if (nRecords > 0 && recordsRead>=nRecords){
			    return false;
			}
			//	check that internal buffer has enough data to read data size
			if (LEN_SIZE_SPECIFIER > dataBuffer.remaining()){
			    reloadBuffer(LEN_SIZE_SPECIFIER);
			    if(dataBuffer.remaining() == 0) return false;
			}
			recordSize = ByteBufferUtils.decodeLength(dataBuffer);
			
			//	check that internal buffer has enough data to read data record
			if (recordSize > dataBuffer.remaining()){
			    reloadBuffer(recordSize);
			    if(recordSize > dataBuffer.remaining()) return false;
			}
			int oldLimit = dataBuffer.limit();
            dataBuffer.mark(); //could be used later in reget method
			dataBuffer.limit(dataBuffer.position() + recordSize);
            recordBuffer.clear();
			recordBuffer.put(dataBuffer);
			recordBuffer.flip();
			dataBuffer.limit(oldLimit);
			
			recordsRead++;
			return true;
		}

         /**
          * Read again previously read record. Use caution when calling this
          * method as it lifts most of the safety checks. Should be used
          * only immediately after successful get(ByteBuffer) call
          * 
         * @param recordBuffer
         * @return
         * @throws IOException
         * @since 27.2.2007
         */
        boolean reget(CloverBuffer recordBuffer) {
                dataBuffer.reset();
                int oldLimit = dataBuffer.limit();
                dataBuffer.limit(dataBuffer.position() + recordSize);
                recordBuffer.clear();
                recordBuffer.put(dataBuffer);
                recordBuffer.flip();
                dataBuffer.limit(oldLimit);
                return true;
            }
         
         
		 /**
		  * Returns next record from the buffer - FIFO order.
		 * @param data	DataRecord into which load the data
		 * @return
		 * @throws IOException
		 * @throws InterruptedException 
		 */
		boolean get(DataRecord data) throws IOException, InterruptedException {
				if(!canRead){
					throw new IOException("Buffer has not been rewind !");
				}
				
				if (nRecords > 0 && recordsRead>=nRecords){
				    return false;
				}
				//	check that internal buffer has enough data to read data size
				if (LEN_SIZE_SPECIFIER > dataBuffer.remaining()) {
				    reloadBuffer(LEN_SIZE_SPECIFIER);
                    if(dataBuffer.remaining() == 0) return false;
				}
                recordSize = ByteBufferUtils.decodeLength(dataBuffer);
				
				//	check that internal buffer has enough data to read data record
				if (recordSize > dataBuffer.remaining()){
				    reloadBuffer(recordSize);
                    if(recordSize > dataBuffer.remaining()) return false;
				}
                dataBuffer.mark();//could be used later for reget
				data.deserialize(dataBuffer);
				
				recordsRead++;
				return true;
			}
		

        boolean reget(DataRecord data) {
            dataBuffer.reset();
            data.deserialize(dataBuffer);
            return true;
        }
        
        
		/**
		 *  Flushes in memory buffer into TMP file
		 *
		 *@exception  IOException  Description of Exception
		 * @throws InterruptedException 
		 *@since                   September 17, 2002
		 */
		private void flushBuffer() throws IOException, InterruptedException {
			try {
				dataBuffer.flip();
				tmpFileChannel.write(dataBuffer.buf());
				dataBuffer.clear();
			} catch (ClosedChannelException e) {
				throw new InterruptedException();
			}
		}
		
		private void reloadBuffer(int requestedSize) throws IOException, InterruptedException {
			try {
				dataBuffer.compact();
				//we have to ensure that the buffer is big enough to bear 'requestedSize' bytes
				if (dataBuffer.capacity() < requestedSize) {
					dataBuffer.expand(0, requestedSize);
				}
				tmpFileChannel.read(dataBuffer.buf());
				dataBuffer.flip();
			} catch (ClosedChannelException e) {
				throw new InterruptedException();
			}
		}

		@Override
		public String toString(){
		    return "start: "+offsetStart+" #records: "+nRecords+" length: "+length;
		}
	}
	
}

